package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSqlApp;
import com.atguigu.gmall.realtime.bean.ProvinceStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static com.atguigu.gmall.realtime.common.Constant.CLICKHOUSE_TABLE_PROVINCE_STATS;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/1 14:43
 */
public class DwsProvince extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsProvince().init(4003, 1, "DwsProvince");
    }
    
    @Override
    protected void run(StreamTableEnvironment tenv) {
        // 1. 创建动态表与source关联:kafka中的topic  订单宽表
        tenv.executeSql("create table order_wide(" +
                            "   province_id bigint, " +
                            "   province_name string, " +
                            "   province_area_code string, " +
                            "   province_iso_code string, " +
                            "   province_3166_2_code string, " +
                            "   order_id bigint, " +
                            "   split_total_amount decimal(20, 2), " +
                            "   create_time string, " +
                            "   et as to_timestamp(create_time), " +
                            "   watermark for et as et - interval '3' second" +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092', " +
                            "   'properties.group.id' = 'DwsProvince', " +
                            "   'topic' = '" + Constant.TOPIC_DWM_ORDER_WIDE + "', " +
                            "   'scan.startup.mode' = 'earliest-offset', " +
                            "   'format' = 'json' " +
                            ")");
        
        /*
         开窗聚合
         */
        
        Table table = tenv.sqlQuery("select " +
                                        "   date_format(tumble_start(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss') stt, " +
                                        "   date_format(tumble_end(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss') edt, " +
                                        "   province_id, " +
                                        "   province_name, " +
                                        "   province_area_code area_code, " +
                                        "   province_iso_code iso_code, " +
                                        "   province_3166_2_code iso_3166_2, " +
                                        "   sum(split_total_amount) order_amount, " +
                                        "   count(distinct(order_id)) order_count," +
                                        "   unix_timestamp() * 1000 ts " +  // 订单宽表粒度是sku, 所以需要对订单id去重再计数
                                        "from order_wide    " +
                                        "group by " +
                                        " province_id," +
                                        " province_name," +
                                        " province_area_code," +
                                        " province_iso_code," +
                                        " province_3166_2_code, " +
                                        " tumble(et, interval '5' second)");
        
        // 把数据写出到clickhouse
        
        tenv
            .toAppendStream(table, ProvinceStats.class)
            .addSink(FlinkSinkUtil
                         .getClickhouseSink(
                             Constant.CLICKHOUSE_DB,
                             CLICKHOUSE_TABLE_PROVINCE_STATS,
                             ProvinceStats.class
                         ));
        
    }
}
